package dao

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"fmt"
	"github.com/go-redis/redis"
	"github.com/jackc/pgconn"
	"github.com/nsqio/go-nsq"
	"github.com/pkg/errors"
	"gorm.io/driver/postgres"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"pay/internal/conf"
	"pay/internal/constant"
	"pay/pkg/cache"
	"pay/pkg/ecode"
	"pay/pkg/env"
	"pay/pkg/log"
	"pay/pkg/mail"
	"pay/pkg/orm"
	"pay/pkg/request_key"
	"strings"
)

type Dao struct {
	db          *gorm.DB
	RedisClient *redis.Client
	NSQProducer *nsq.Producer
	// 邮件执行器
	MailClient *mail.SMTPSender
}

type _transactionKey struct{}
type _nsqTransactionMsgs struct{}

func (d *Dao) Close(ctx context.Context) {
	orm.CloseConnection(ctx, d.db)
	cache.Close(ctx, d.RedisClient)
}

func New(ctx context.Context, dbConfig *orm.Config, redisConfig *cache.Config, env env.Type,
	mailConfig *mail.Config) (dao *Dao, err error) {
	dao = &Dao{}
	if dao.RedisClient, err = cache.New(redisConfig); err != nil {
		return
	}
	log.FromContext(ctx).Infow("redis连接...完成")
	log.FromContext(ctx).Infow("数据库连接创建...开始")
	pg := postgres.Open(dbConfig.DSN)
	if dao.db, err = orm.NewConnection(dbConfig, env, pg); err != nil {
		return
	}
	log.FromContext(ctx).Infow("数据库连接创建...完成")
	// 加载启动配置信息
	if err = dao.ReloadBootConfig(ctx); err != nil {
		return
	}
	if dao.MailClient, err = mailConfig.Init(); err != nil {
		return
	}
	return
}

// Ping verify server is ok.
func (d *Dao) Ping(ctx context.Context) (err error) {
	db, _ := d.db.DB()
	err = db.Ping()
	if err != nil && errors.Is(err, driver.ErrBadConn) {
		err = db.Ping()
	}
	if err != nil {
		log.FromContext(ctx).Infof("%s dao.db.Ping() error(%v)", d.db.Dialector.Name(), err)
		err = ecode.WithStack(err)
	}
	return
}

// 数据库事务只附加一次上下文
func (d *Dao) TransContext(ctx context.Context) *gorm.DB {
	db, ok := ctx.Value(_transactionKey{}).(*gorm.DB)
	if !ok {
		db = d.db.WithContext(ctx)
	}
	return db.WithContext(ctx)
}

func (d *Dao) processSqlError(err error) error {
	if err == nil {
		return nil
	}
	if pgError, ok := err.(*pgconn.PgError); ok {
		if pgError.SQLState() == "23505" {
			err = constant.ErrDuplicateInput
		} else if pgError.SQLState() == "22003" {
			err = constant.ErrMaximumNumber
		}
	} else if strings.Contains(err.Error(), "is greater than maximum value for") {
		err = constant.ErrMaximumNumber
	}
	return ecode.WithStack(err)
}

// Create 通用的创建方法
func (d *Dao) Create(ctx context.Context, value interface{}) error {
	return d.processSqlError(d.TransContext(ctx).Create(value).Error)
}

// Update 通用的更新方法
func (d *Dao) Update(ctx context.Context, m, c interface{}) error {
	return d.processSqlError(d.TransContext(ctx).Model(m).Updates(c).Error)
}

// Create 通用的创建方法
func (d *Dao) Save(ctx context.Context, c interface{}) error {
	return d.processSqlError(d.TransContext(ctx).Save(c).Error)
}

// 自动基于打gorm primary tag的主键更新
func (d *Dao) AutoUpdate(ctx context.Context, val interface{}) error {
	query, args, updateMap := orm.FormatColumnMap(val)
	return d.processSqlError(d.TransContext(ctx).Model(val).Where(
		query, args...).Updates(updateMap).Error)
}

// Lock 锁定单条记录
func (d *Dao) LockObjectByID(ctx context.Context, rcv interface{}, id interface{}, sgId, pStoreId uint32) error {
	return d.processSqlError(d.TransContext(ctx).Clauses(
		clause.Locking{Strength: "UPDATE"}).Where(
		"id = ? AND sg_id = ? AND p_store_id = ? AND status = ?",
		id, sgId, pStoreId, true).Take(rcv).Error)
}

// Lock 锁定单条记录
func (d *Dao) LockTradeByOrderNumber(ctx context.Context, rcv interface{}, orderNumber string) error {
	return d.processSqlError(d.TransContext(ctx).Clauses(
		clause.Locking{Strength: "UPDATE"}).Where(
		"order_number = ?", orderNumber).Take(rcv).Error)
}

// Lock 锁定单条记录
func (d *Dao) LockTradeByOrderID(ctx context.Context, rcv interface{}, orderId uint64) error {
	return d.processSqlError(d.TransContext(ctx).Clauses(
		clause.Locking{Strength: "UPDATE"}).Where(
		"id = ?", orderId).Take(rcv).Error)
}

// Lock 锁定单条记录
func (d *Dao) LockObjectByName(ctx context.Context, rcv interface{}, name string, sgId, pStoreId uint32) error {
	return d.processSqlError(d.TransContext(ctx).Clauses(
		clause.Locking{Strength: "UPDATE"}).Where(
		"sg_id = ? AND p_store_id = ? AND name = ? AND status = ?",
		sgId, pStoreId, name, true).Take(rcv).Error)
}

// Execute 通用的SQL执行器
func (d *Dao) Execute(ctx context.Context, sql string, value ...interface{}) error {
	err := d.TransContext(ctx).Exec(sql, value...).Error
	if err != nil && errors.Is(err, driver.ErrBadConn) {
		// 重试一次，防止bad connection
		err = d.TransContext(ctx).Create(value).Error
	}
	return d.processSqlError(err)
}

// Transaction 通过ctx传递事务
func (d *Dao) Transaction(ctx context.Context, fc func(ctx context.Context) error, opts ...*sql.TxOptions) (err error) {
	// 清空NSQ消息
	reqID := request_key.GetRequestID(ctx)
	if len(reqID) == 0 {
		err = fmt.Errorf("no request_id found in transaction")
		return
	}
	tranCtx := context.WithValue(ctx, _nsqTransactionMsgs{}, make(map[string]map[uint64][]byte))
	if err = d.db.Transaction(func(tx *gorm.DB) (err error) {
		tranCtx = context.WithValue(tranCtx, _transactionKey{}, tx)
		return fc(tranCtx)
	}, opts...); err != nil {
		return
	}
	if nsqTransactionMsgs, ok := tranCtx.Value(_nsqTransactionMsgs{}).(map[string]map[uint64][]byte); ok {
		if len(nsqTransactionMsgs) != 0 {
			// 事务成功后，发送NSQ异步消息
			for topic, msgMap := range nsqTransactionMsgs {
				nsqChannel := make(chan *nsq.ProducerTransaction)
				msgIds := make([]uint64, 0)
				msgBodies := make([][]byte, 0)
				for msgId, msgBody := range msgMap {
					msgIds = append(msgIds, msgId)
					msgBodies = append(msgBodies, msgBody)
				}
				go func() {
					newCtx := request_key.GenerateContextWithRequestID(reqID)
					defer ecode.AsyncFailOver(newCtx, "数据库事务消息发送", conf.Conf.Env, nil)
					res := <-nsqChannel
					if res.Error != nil {
						d.nsqFailed(newCtx, topic, msgIds, msgBodies, res.Error)
					} else {
						d.nsqSucceeded(newCtx, topic, msgIds, msgBodies)
					}
				}()
				if err = d.NSQProducer.MultiPublishAsync(topic, msgBodies, nsqChannel); err != nil {
					d.nsqFailed(ctx, topic, msgIds, msgBodies, err)
					// NSQ发送失败时，不抛出错误
					err = nil
				}
			}
		}
	} else {
		err = fmt.Errorf("invalid nsq trans msgs")
		return
	}
	return
}

// Count 通用检查数量的方法
func (d *Dao) Count(ctx context.Context, m, c interface{}) (count int64, err error) {
	err = d.TransContext(ctx).Model(m).Where(c).Count(&count).Error
	return count, d.processSqlError(err)
}

// CheckExist 通用检查是否存在的方法
func (d *Dao) CheckExist(ctx context.Context, m, c interface{}) (isExist bool, err error) {
	var count int64
	count, err = d.Count(ctx, m, c)
	return count >= 1, d.processSqlError(err)
}
